use crate::block::BlockContents;
use crate::block_builder::BlockBuilder;
use crate::blockhandle::BlockHandle;
use crate::cmp::InternalKeyCmp;
use crate::compressor::{self, Compressor, CompressorId};
use crate::crc;
use crate::error::Result;
use crate::filter::{InternalFilterPolicy, NoFilterPolicy};
use crate::filter_block::FilterBlockBuilder;
use crate::key_types::InternalKey;
use crate::log::mask_crc;
use crate::options::Options;

use std::cmp::Ordering;
use std::io::Write;
use std::rc::Rc;

use integer_encoding::FixedIntWriter;

pub const FOOTER_LENGTH: usize = 40;
pub const FULL_FOOTER_LENGTH: usize = FOOTER_LENGTH + 8;
pub const MAGIC_FOOTER_NUMBER: u64 = 0xdb4775248b80fb57;
pub const MAGIC_FOOTER_ENCODED: [u8; 8] = [0x57, 0xfb, 0x80, 0x8b, 0x24, 0x75, 0x47, 0xdb];

pub const TABLE_BLOCK_COMPRESS_LEN: usize = 1;
pub const TABLE_BLOCK_CKSUM_LEN: usize = 4;

/// Footer is a helper for encoding/decoding a table footer.
#[derive(Debug, Clone)]
pub struct Footer {
    pub meta_index: BlockHandle,
    pub index: BlockHandle,
}

/// A Table footer contains a pointer to the metaindex block, another pointer to the index block,
/// and a magic number:
/// [ { table data ... , METAINDEX blockhandle, INDEX blockhandle, PADDING bytes } = 40 bytes,
/// MAGIC_FOOTER_ENCODED ]
impl Footer {
    pub fn new(metaix: BlockHandle, index: BlockHandle) -> Footer {
        Footer {
            meta_index: metaix,
            index,
        }
    }

    pub fn decode(from: &[u8]) -> Option<Footer> {
        assert!(from.len() >= FULL_FOOTER_LENGTH);
        assert_eq!(&from[FOOTER_LENGTH..], &MAGIC_FOOTER_ENCODED);
        let (meta, metalen) = BlockHandle::decode(&from[0..])?;
        let (ix, _) = BlockHandle::decode(&from[metalen..])?;

        Some(Footer {
            meta_index: meta,
            index: ix,
        })
    }

    pub fn encode(&self, to: &mut [u8]) {
        assert!(to.len() >= FOOTER_LENGTH + 8);

        let s1 = self.meta_index.encode_to(to);
        let s2 = self.index.encode_to(&mut to[s1..]);

        (s1 + s2..FOOTER_LENGTH).for_each(|i| {
            to[i] = 0;
        });
        (FOOTER_LENGTH..FULL_FOOTER_LENGTH).for_each(|i| {
            to[i] = MAGIC_FOOTER_ENCODED[i - FOOTER_LENGTH];
        });
    }
}

/// A table consists of DATA BLOCKs, META BLOCKs, a METAINDEX BLOCK, an INDEX BLOCK and a FOOTER.
///
/// DATA BLOCKs, META BLOCKs, INDEX BLOCK and METAINDEX BLOCK are built using the code in
/// the `block` module.
///
/// The FOOTER consists of a BlockHandle that points to the metaindex block, another pointing to
/// the index block, padding to fill up to 40 B and at the end the 8B magic number
/// 0xdb4775248b80fb57.
///
/// 这段代码定义了 TableBuilder 结构体，它是 LSM 树存储引擎中用于构建 SSTable（排序字符串表）文件的构造器，负责将键值对数据按照特定格式组织并写入到输出流中。
///
/// SSTable 是 LSM 树中持久化存储数据的关键结构，TableBuilder 按照固定格式逐步构建这种文件，
/// 其整体结构包括：数据块（DATA BLOCKs）、元数据块（META BLOCKs）、元索引块（METAINDEX BLOCK）、索引块（INDEX BLOCK）和页脚（FOOTER）。
///
/// TableBuilder 的典型使用流程包括：
///
/// 逐步添加键值对到 data_block，当数据块达到预设大小时，将其写入输出流（dst），并记录其偏移量和最后一个键。
///
/// 用各数据块的偏移量和最后一个键构建 index_block，形成数据块的索引。
///
/// 同时通过 filter_block 为键构建过滤器（如布隆过滤器），优化查询性能。
///
/// 所有数据块写入完成后，依次写入元索引块、索引块和页脚（FOOTER），其中页脚包含元索引块和索引块的位置信息及魔术数（用于验证文件完整性）。
///
/// 通过这种结构化的构建过程，TableBuilder 确保最终生成的 SSTable 符合格式规范，既能高效存储键值对，又能支持快速查询和范围遍历。
pub struct TableBuilder<Dst: Write> {
    /// 数据库配置选项，包含构建 SSTable 所需的参数（如块大小、压缩方式、过滤器策略等）。
    opt: Options,
    /// 输出流（实现 Write 特性），用于将构建好的 SSTable 数据写入到磁盘文件或其他存储目标。
    dst: Dst,

    /// 当前写入位置在输出流中的偏移量，用于跟踪各块在文件中的位置。
    offset: usize,
    /// 记录已写入 SSTable 的键值对总数。
    num_entries: usize,
    /// 存储上一个数据块的最后一个键，用于构建索引块（INDEX BLOCK）时记录各数据块的键范围。
    prev_block_last_key: Vec<u8>,

    /// 数据块构造器，用于组装键值对数据块（实际存储键值对的地方）。
    data_block: Option<BlockBuilder>,
    /// 索引块构造器，用于记录各数据块的偏移量和键范围（便于快速定位数据块）。
    index_block: Option<BlockBuilder>,
    /// 过滤器块构造器，用于构建布隆过滤器等结构（加速键存在性检查）。
    filter_block: Option<FilterBlockBuilder>,
}

impl<Dst: Write> TableBuilder<Dst> {
    pub fn new_no_filter(mut opt: Options, dst: Dst) -> TableBuilder<Dst> {
        opt.filter_policy = Rc::new(Box::new(NoFilterPolicy::new()));
        TableBuilder::new(opt, dst)
    }
}

/// TableBuilder is used for building a new SSTable. It groups entries into blocks,
/// calculating checksums and bloom filters.
impl<Dst: Write> TableBuilder<Dst> {
    /// Create a new table builder.
    /// The comparator in opt will be wrapped in a InternalKeyCmp, and the filter policy
    /// in an InternalFilterPolicy.
    pub fn new(mut opt: Options, dst: Dst) -> TableBuilder<Dst> {
        opt.cmp = Rc::new(Box::new(InternalKeyCmp(opt.cmp.clone())));
        opt.filter_policy = Rc::new(Box::new(InternalFilterPolicy::new(opt.filter_policy)));
        TableBuilder::new_raw(opt, dst)
    }

    /// Like new(), but doesn't wrap the comparator in an InternalKeyCmp (for testing)
    pub fn new_raw(opt: Options, dst: Dst) -> TableBuilder<Dst> {
        TableBuilder {
            opt: opt.clone(),
            dst,
            offset: 0,
            prev_block_last_key: vec![],
            num_entries: 0,
            data_block: Some(BlockBuilder::new(opt.clone())),
            filter_block: Some(FilterBlockBuilder::new(opt.filter_policy.clone())),
            index_block: Some(BlockBuilder::new(opt)),
        }
    }

    pub fn entries(&self) -> usize {
        self.num_entries
    }

    /// 这段代码定义了 size_estimate 方法，用于估算某个数据结构（很可能是 SSTable 文件）的总大小，通常用于存储引擎中的空间管理、容量计算或性能优化。
    ///
    /// 这个方法的作用是快速估算复杂结构的总大小，而无需实际计算精确大小（可能涉及大量 IO 或复杂计算）。在存储引擎中，它可用于：
    ///
    /// 判断 SSTable 是否达到合并阈值（compaction trigger）。
    ///
    /// 管理缓存策略（如根据大小决定是否缓存整个结构）。
    ///
    /// 统计磁盘使用量或预分配存储空间。
    ///
    /// 简单来说，它是一个 “空间计算器”，用于快速得到某个数据结构（如 SSTable）的总大小估算值，辅助存储引擎的各种空间管理决策。
    pub fn size_estimate(&self) -> usize {
        // 累加各组成部分的大小：
        // 分别检查并累加 data_block（数据块）、index_block（索引块）、filter_block（过滤器块）的估算大小（通过各自的 size_estimate() 方法）。
        // 这些块是 SSTable 的核心组成部分：数据块存储实际键值对，索引块存储数据块的索引信息，过滤器块存储布隆过滤器等快速查找结构。
        let mut size = 0;
        if let Some(ref b) = self.data_block {
            size += b.size_estimate();
        }
        if let Some(ref b) = self.index_block {
            size += b.size_estimate();
        }
        if let Some(ref b) = self.filter_block {
            size += b.size_estimate();
        }
        // self.offset：可能是指某个偏移量相关的存储开销（如块之间的间隔、元数据偏移等）。
        // FULL_FOOTER_LENGTH：SSTable 文件末尾的固定长度 Footer 开销（通常包含索引块偏移量、校验和等元信息，长度固定）。
        // 返回总估算大小：所有部分的大小之和即为该结构的总估算大小（单位通常是字节）。
        size + self.offset + FULL_FOOTER_LENGTH
    }

    /// Add a key to the table. The key as to be lexically greater or equal to the last one added.
    ///
    /// 这段代码定义了 add 方法，用于向 SSTable（通过 TableBuilder 构建）中添加键值对，是构建 SSTable 过程中的核心写入操作。它确保了数据按照正确的格式和顺序被写入，并处理了数据块的分割逻辑。
    ///
    /// 这个方法的核心作用是保证键值对按照有序方式写入 SSTable，并在适当的时候分割数据块，既维护了 SSTable 的有序性（查询基础），又控制了数据块大小（性能优化）。
    ///
    /// 在实际构建 SSTable 时，调用方会逐条调用 add 方法写入键值对，而 add 方法内部自动处理数据块的分割和过滤器更新，最终确保生成的 SSTable 符合格式规范且性能最优。
    pub fn add(&mut self, key: InternalKey<'_>, val: &[u8]) -> Result<()> {
        // 确保数据块构造器已初始化，避免在未准备好的状态下添加数据。
        assert!(self.data_block.is_some());
        // 若存在上一个数据块的最后一个键（prev_block_last_key），则通过比较器（self.opt.cmp）验证当前键是否大于上一个键（Ordering::Less），
        // 确保 SSTable 中的键是严格递增的（这是 SSTable 的核心特性，保证查询效率）。
        if !self.prev_block_last_key.is_empty() {
            assert!(self.opt.cmp.cmp(&self.prev_block_last_key, key) == Ordering::Less);
        }
        // 检查当前数据块（data_block）的估算大小是否超过配置的块大小（self.opt.block_size）。
        // 若超过，则调用 self.write_data_block(key) 完成当前数据块的写入，并创建新的数据块（确保每个数据块不超过预设大小，优化读写性能）。
        if self.data_block.as_ref().unwrap().size_estimate() > self.opt.block_size {
            self.write_data_block(key)?;
        }
        // 若启用了过滤器（filter_block 存在），则将当前键添加到过滤器中（如布隆过滤器），用于后续快速判断键是否存在于该 SSTable 中。
        let dblock = &mut self.data_block.as_mut().unwrap();

        if let Some(ref mut fblock) = self.filter_block {
            fblock.add_key(key);
        }
        // 递增键值对计数器（num_entries）。
        // 通过数据块构造器（dblock.add）将当前键值对添加到数据块中（内部会处理键的前缀压缩等存储优化）。
        self.num_entries += 1;
        dblock.add(key, val);
        Ok(())
    }

    /// Writes an index entry for the current data_block where `next_key` is the first key of the next block.
    /// Calls write_block() for writing the block to disk.
    ///
    /// 这段代码定义了 write_data_block 方法，是 TableBuilder 构建 SSTable 过程中处理数据块写入的核心逻辑，负责将内存中组装好的数据块持久化到磁盘，并为其创建索引条目，确保后续能快速定位该数据块。
    ///
    /// 这个方法的核心作用是将内存中的数据块 “落地” 到磁盘，并为其建立索引，是 SSTable 从内存结构到磁盘文件的关键转换步骤。
    /// 它通过生成分隔符和索引条目，确保后续查询时能快速定位到目标键所在的数据块，同时通过压缩减少磁盘占用。
    fn write_data_block(&mut self, next_key: InternalKey<'_>) -> Result<()> {
        // 断言确保数据块构造器（data_block）已初始化。
        assert!(self.data_block.is_some());
        // 通过 take().unwrap() 取出当前组装好的数据块（block），并将其从 TableBuilder 中移除（后续会创建新的数据块）。
        let block = self.data_block.take().unwrap();
        // 调用 find_shortest_sep 生成当前数据块最后一个键与下一个数据块第一个键（next_key）之间的最短分隔符（sep）。
        // 这个分隔符用于索引块，标识当前数据块的键范围上限（小于该分隔符的键都在当前数据块中）。
        let sep = self.opt.cmp.find_shortest_sep(block.last_key(), next_key);
        // 保存当前数据块的最后一个键到 prev_block_last_key，用于后续验证键的有序性。
        self.prev_block_last_key = Vec::from(block.last_key());
        // 完成数据块的构建（block.finish()），得到其二进制内容（contents）。
        let contents = block.finish();
        // 使用配置的压缩器（compressor）压缩数据块，并通过 write_block 方法将其写入输出流（dst），返回该数据块在文件中的位置信息（handle，包含偏移量和大小）。
        let compressor_list = self.opt.compressor_list.clone();
        let compressor = compressor_list.get(self.opt.compressor)?;
        // 写入到数据文件
        let handle = self.write_block(contents, (self.opt.compressor, compressor))?;
        // 将数据块的位置信息（handle）编码为字节数组（handle_enc）。
        let mut handle_enc = [0_u8; 16];
        let enc_len = handle.encode_to(&mut handle_enc);
        // 向索引块（index_block）中添加条目：键为分隔符（sep），值为编码后的位置信息，形成数据块的索引。
        self.index_block
            .as_mut()
            .unwrap()
            .add(&sep, &handle_enc[0..enc_len]);
        // 创建新的数据块构造器（BlockBuilder），赋值给 data_block，用于接收后续键值对。
        self.data_block = Some(BlockBuilder::new(self.opt.clone()));
        // 若启用了过滤器块（filter_block），调用 start_block 记录当前数据块的偏移量，确保过滤器能关联到正确的数据块。
        if let Some(ref mut fblock) = self.filter_block {
            fblock.start_block(self.offset);
        }

        Ok(())
    }

    /// Calculates the checksum, writes the block to disk and updates the offset.
    ///
    /// 这段代码定义了 write_block 方法，用于将 SSTable 中的块（可以是数据块、索引块、过滤器块等）压缩、计算校验和并写入磁盘，同时记录块在文件中的位置信息，是 SSTable 持久化过程中的关键底层操作。
    ///
    /// 这个方法的核心作用是标准化块的持久化流程，确保所有块（数据块、索引块等）都以统一格式写入磁盘，同时通过压缩节省空间、通过校验和保证数据可靠性。
    /// 它是 SSTable 底层存储格式的执行者，直接影响文件的存储效率和数据完整性。
    ///
    /// 简单来说，它是块数据的 “磁盘写入器”，负责将内存中的块安全、高效地持久化到磁盘，并返回其位置信息供后续访问。
    fn write_block(
        &mut self,
        block: BlockContents,
        compressor_id_pair: (u8, &dyn Compressor),
    ) -> Result<BlockHandle> {
        // 接收待写入的块内容（block: BlockContents）和压缩器信息（compressor_id_pair，包含压缩类型 ID 和压缩器实例）。
        let (ctype, compressor) = compressor_id_pair;
        // 使用指定的压缩器（compressor.encode）对块数据进行压缩，得到压缩后的数据（data），目的是减少磁盘存储空间占用。
        let data = compressor.encode(block.to_vec())?;
        // 初始化 CRC 校验和计算器（digest）。
        let mut digest = crc::digest();
        // 对压缩后的数据（data）和压缩类型标识（[ctype; TABLE_BLOCK_COMPRESS_LEN]）计算校验和，用于后续验证数据完整性（检测磁盘损坏或传输错误）。
        digest.update(&data);
        digest.update(&[ctype; TABLE_BLOCK_COMPRESS_LEN]);
        // 将压缩后的数据（data）、压缩类型标识（固定长度的 ctype 数组）、校验和（通过 write_fixedint 写入）依次写入输出流（self.dst）。
        // 这三部分共同构成磁盘上的块存储格式：[压缩数据][压缩类型][校验和]。
        self.dst.write_all(&data)?;
        self.dst.write_all(&[ctype; TABLE_BLOCK_COMPRESS_LEN])?;
        self.dst.write_fixedint(mask_crc(digest.finalize()))?;
        // 创建 BlockHandle 结构体，记录当前块的起始偏移量（self.offset）和压缩后的数据长度（data.len()），用于后续快速定位该块。
        let handle = BlockHandle::new(self.offset, data.len());
        // 更新当前写入偏移量（self.offset），加上压缩数据长度、压缩类型标识长度（TABLE_BLOCK_COMPRESS_LEN）和校验和长度（TABLE_BLOCK_CKSUM_LEN），确保下一次写入从正确位置开始。
        self.offset += data.len() + TABLE_BLOCK_COMPRESS_LEN + TABLE_BLOCK_CKSUM_LEN;
        // 返回 BlockHandle，供上层逻辑（如索引块构建）记录块的位置信息。
        Ok(handle)
    }

    /// 这段代码定义了 finish 方法，用于完成 SSTable 的构建并写入最终的元数据和页脚，是 TableBuilder 构建 SSTable 过程的收尾操作。
    /// 它确保所有数据块、索引和元数据都被正确写入，并生成符合格式规范的 SSTable 文件。
    ///
    /// 这个方法的核心作用是将 SSTable 的所有组成部分（数据块、过滤器块、元索引块、索引块）按格式组装并写入最终文件，是 SSTable 从内存结构持久化为磁盘文件的关键步骤。
    ///
    /// 在实际流程中，finish 被调用后，TableBuilder 会生成一个完整的、可被存储引擎读取和查询的 SSTable 文件，包含所有键值对数据及高效查询所需的元数据结构。
    ///
    /// 简单来说，它是 SSTable 构建的 “收尾器”，负责将所有零散的块整合成一个完整的、符合规范的 SSTable 文件。
    pub fn finish(mut self) -> Result<usize> {
        assert!(self.data_block.is_some());
        // 获取配置的压缩器列表（compressor_list）和当前使用的压缩器（compressor_id_pair），用于后续块的压缩处理（如数据块、索引块可能需要压缩以节省空间）。
        let compressor_list = self.opt.compressor_list.clone();
        let compressor_id_pair = (
            self.opt.compressor,
            compressor_list.get(self.opt.compressor)?,
        );

        // If there's a pending data block, write it
        // 如果数据块（data_block）中仍有未写入的键值对，调用 write_data_block 完成其写入。
        if self.data_block.as_ref().unwrap().entries() > 0 {
            // Find a key reliably past the last key
            // 通过 find_short_succ 生成一个比最后一个键大的 “哨兵键”（key_past_last），确保索引块能正确标记该数据块的键范围。
            let key_past_last = self
                .opt
                .cmp
                .find_short_succ(self.data_block.as_ref().unwrap().last_key());
            self.write_data_block(&key_past_last)?;
        }

        // Create metaindex block
        // 初始化元索引块构造器（meta_ix_block）。
        let mut meta_ix_block = BlockBuilder::new(self.opt.clone());
        // 若存在过滤器块（filter_block）：
        if self.filter_block.is_some() {
            // if there's a filter block, write the filter block and add it to the metaindex block.
            let fblock = self.filter_block.take().unwrap();
            let filter_key = format!("filter.{}", fblock.filter_name());
            let fblock_data = fblock.finish();
            // 完成过滤器块的构建并写入文件，获取其在文件中的位置信息（fblock_handle）。
            let fblock_handle = self.write_block(
                fblock_data.into(),
                (compressor::NoneCompressor::ID, &compressor::NoneCompressor),
            )?;

            let mut handle_enc = [0_u8; 16];
            let enc_len = fblock_handle.encode_to(&mut handle_enc);
            // 将过滤器块的名称（如 filter.bloom）和位置信息编码后添加到元索引块中，使后续能快速找到过滤器。
            meta_ix_block.add(filter_key.as_bytes(), &handle_enc[0..enc_len]);
        }

        // write metaindex block
        // 完成元索引块的构建并写入文件，记录其位置（meta_ix_handle）。
        let meta_ix = meta_ix_block.finish();
        let meta_ix_handle = self.write_block(meta_ix, compressor_id_pair)?;

        // write index block
        // 完成索引块（index_block）的构建并写入文件，记录其位置（ix_handle）。索引块存储了所有数据块的位置和键范围，是查询的入口。
        let index_cont = self.index_block.take().unwrap().finish();
        let ix_handle = self.write_block(index_cont, compressor_id_pair)?;

        // write footer.
        // 构建页脚（Footer），包含元索引块和索引块的位置信息。
        let footer = Footer::new(meta_ix_handle, ix_handle);
        // 将页脚编码后写入文件末尾（固定长度为 FULL_FOOTER_LENGTH），并包含魔术数（用于验证文件完整性）。
        let mut buf = [0; FULL_FOOTER_LENGTH];
        footer.encode(&mut buf);

        self.offset += self.dst.write(&buf[..])?;
        // 刷新输出流（dst.flush()）确保所有数据写入磁盘。
        self.dst.flush()?;
        // 返回文件总大小（self.offset），即构建完成的 SSTable 的总字节数。
        Ok(self.offset)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::blockhandle::BlockHandle;
    use crate::options;

    #[test]
    fn test_footer() {
        let f = Footer::new(BlockHandle::new(44, 4), BlockHandle::new(55, 5));
        let mut buf = [0; 48];
        f.encode(&mut buf[..]);

        let f2 = Footer::decode(&buf).unwrap();
        assert_eq!(f2.meta_index.offset(), 44);
        assert_eq!(f2.meta_index.size(), 4);
        assert_eq!(f2.index.offset(), 55);
        assert_eq!(f2.index.size(), 5);
    }

    #[test]
    fn test_table_builder() {
        let mut d = Vec::with_capacity(512);
        let mut opt = options::for_test();
        opt.block_restart_interval = 3;
        opt.compressor = compressor::SnappyCompressor::ID;
        let mut b = TableBuilder::new_raw(opt, &mut d);

        let data = [
            ("abc", "def"),
            ("abe", "dee"),
            ("bcd", "asa"),
            ("dcc", "a00"),
        ];
        let data2 = [
            ("abd", "def"),
            ("abf", "dee"),
            ("ccd", "asa"),
            ("dcd", "a00"),
        ];

        for i in 0..data.len() {
            b.add(data[i].0.as_bytes(), data[i].1.as_bytes()).unwrap();
            b.add(data2[i].0.as_bytes(), data2[i].1.as_bytes()).unwrap();
        }

        let estimate = b.size_estimate();

        assert_eq!(143, estimate);
        assert!(b.filter_block.is_some());

        let actual = b.finish().unwrap();
        assert_eq!(223, actual);
    }

    #[test]
    #[should_panic]
    fn test_bad_input() {
        let mut d = Vec::with_capacity(512);
        let mut opt = options::for_test();
        opt.block_restart_interval = 3;
        let mut b = TableBuilder::new_raw(opt, &mut d);

        // Test two equal consecutive keys
        let data = [
            ("abc", "def"),
            ("abc", "dee"),
            ("bcd", "asa"),
            ("bsr", "a00"),
        ];

        for &(k, v) in data.iter() {
            b.add(k.as_bytes(), v.as_bytes()).unwrap();
        }
        b.finish().unwrap();
    }
}
